package com.oracle.calllog.analyze.mapper;

import java.io.IOException;

import com.oracle.calllog.analyze.util.Constants;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;

/*
 * 统计每个人：每年-每年中的每月-每年中每月中的每天
 * 			的通话次数和通话时长
 * Mapper的输出结构：联系人_日期
 * 1888888888_2021		duration
 * 1888888888_202101		duration
 * 1888888888_20210109	...
 * 1777777777_2021
 * 1777777777_202101
 * 1777777777_20210109
 */
public class CallLogAnalyzeMapper extends TableMapper<Text, Text> {
//    private ComboKey comboKey = new ComboKey();
    private Text comboKey = new Text();
    private Text durationValue = new Text();

  

    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        //将一条记录中所记录的数据提取出来
        Cell[] cells = value.rawCells();

        //RowKey的格式为：regionCode_contact1No_buildTime_contact2No_duration_flag
        String rowKey = Bytes.toString(value.getRow());
        String[] datas = rowKey.split("_");

        //基于HBase的一分为二的存储模型，全部联系人做为主叫人和被叫人的数据都是一样的处理流程
        String contact1No = datas[1];
        String contact2No = datas[3];
        String buildTime = datas[2];
        String duration = datas[4];
        String callFlag = datas[5];

        String year = buildTime.substring(0,4);
        String month = buildTime.substring(4,6);
        String day = buildTime.substring(6,8);

        String contact1Name = "";
        for(Cell cell : cells) {
            String qualifierName = Bytes.toString(CellUtil.cloneQualifier(cell));
            //如果是第1个联系人的姓名字段
            if(Constants.QUALIFIER_CONTACT1_NAME.equals(qualifierName)){
                contact1Name = Bytes.toString(CellUtil.cloneValue(cell));
            }
        }

//        //组装DateInfo维度对象
//        DateInfo yearDim = new DateInfo(year, "-1", "-1");
//        DateInfo monthDim = new DateInfo(year, month, "-1");
//        DateInfo dayDim = new DateInfo(year, month, day);
//
//        //组装Contact
//        Contact contact = new Contact(contact1No, contact1Name);

        /***************************************************
         * 以下为基于一条通话日志信息分解出的<B>联系人(包含主叫人和被叫人)</B>的三种时间维度用于统计分析的数据
         ***************************************************/
        // 1 联系人以年为维度单位的呼叫数据整合(这里有个问题：使用Bean对象报ClassNotFoundException错误)
        //年维度的数据示例:	18888888888	张飞	2021年    -1月	-1日		->	1288
//        //组装ComboKey
//        //向组合Key中组装联系人对象
//        comboKey.setContact(contact);
//        //向组合Key中组装【年维度】对象
//        comboKey.setDateInfo(yearDim);
//        //Value为通话时长
//        durationValue.set(duration);
//        //将年维度数据写入下一环节
//        context.write(comboKey, durationValue);

        // 1 联系人以年为维度单位的呼叫数据整合(改为使用字符串组合)
        //年维度的数据示例:	18888888888	张飞	2021年    -1月	-1日		->	1288
        //Key为联系人信息+时间维度
        comboKey.set(contact1No+"|"+contact1Name+"|"+year+"|-1|-1");
        //Value为通话时长
        durationValue.set(duration);
        //将年维度数据写入下一环节
        context.write(comboKey, durationValue);

        // 2 联系人的以月为维度单位的呼叫数据整合(这里有个问题：使用Bean对象报ClassNotFoundException错误)
        //月维度的数据示例:	18888888888	张飞	2021年    3月	-1日		->	1288
        //只需要在年维度数据的基础上改变为月维度即可实现,其它数据不需要调整
//        //向组合Key中组装【月维度】对象
//        comboKey.setDateInfo(monthDim);
//        //月维度数据写入下一环节
//        context.write(comboKey, durationValue);

        // 2 联系人的以月为维度单位的呼叫数据整合(改为使用字符串组合)
        //月维度的数据示例:	18888888888	张飞	2021年    3月	-1日		->	1288
        //Key为联系人信息+时间维度
        comboKey.set(contact1No+"|"+contact1Name+"|"+year+"|"+month+"|-1");
        //Value为通话时长
        durationValue.set(duration);
        //月维度数据写入下一环节
        context.write(comboKey, durationValue);

        
        // 3 联系人的以日为维度单位的呼叫数据整合(这里有个问题：使用Bean对象报ClassNotFoundException错误)
        //日维度的数据示例:	18888888888	张飞	2021年    3月	12日		->	1288
        //只需要在月维度数据的基础上改变为日维度即可实现,其它数据不需要调整
//        //向组合Key中组装【日维度】对象
//        comboKey.setDateInfo(dayDim);
//        //日维度数据写入下一环节
//        context.write(comboKey, durationValue);

        // 3 联系人的以日为维度单位的呼叫数据整合(改为使用字符串组合)
        //日维度的数据示例:	18888888888	张飞	2021年    3月	12日		->	1288
        //Key为联系人信息+时间维度
        comboKey.set(contact1No+"|"+contact1Name+"|"+year+"|"+month+"|"+day);
        //Value为通话时长
        durationValue.set(duration);
        //日维度数据写入下一环节
        context.write(comboKey, durationValue);
    }
}